package com.explorex.infra.puma.gateway.netty.common;

import com.explorex.infra.puma.gateway.netty.common.concurrent.SmartRunnable;
import com.explorex.infra.puma.gateway.netty.rrmp.RrmpMessage;
import com.explorex.infra.puma.gateway.netty.ws.message.Message;
import com.google.common.collect.Sets;

import io.netty.channel.ChannelFutureListener;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;

import java.nio.channels.ClosedChannelException;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;

@Slf4j
public class TopicManager {

    private static final ScheduledExecutorService scheduledExecutor = new ScheduledThreadPoolExecutor(1
            , new BasicThreadFactory.Builder().namingPattern(ChannelManager.class.getSimpleName() + "-%d").build());

    private static final ExecutorService topicPoolExecutor =  Executors.newCachedThreadPool();

    /**
     * topic to channel list
     */
    private final static ConcurrentHashMap<String, Set<BaseSocketSession>> topicsChannel = new ConcurrentHashMap<>();

    private final static ConcurrentHashMap<BaseSocketSession, Set<String>> channelTopics = new ConcurrentHashMap<>();


    static {
    }

    public static void removeTopic(String topic){
        topicsChannel.remove(topic);
    }

    public static void subscribeTopic(String topic, BaseSocketSession baseSocketSession) {
        Set<BaseSocketSession> orDefault = topicsChannel.getOrDefault(topic, Sets.newCopyOnWriteArraySet());
        orDefault.add(baseSocketSession);
        topicsChannel.put(topic, orDefault);

        Set<String> topicsOrDefault = channelTopics.getOrDefault(baseSocketSession, Sets.newCopyOnWriteArraySet());
        topicsOrDefault.add(topic);
        channelTopics.put(baseSocketSession,topicsOrDefault);

        //clear old topics， 此处暂且一个 session 只能在一个topic 之内, 后续扩展再说
        Set<String> oldTopics = channelTopics.get(baseSocketSession);
        if (CollectionUtils.isNotEmpty(oldTopics)){
            oldTopics.stream().filter(ele->!topic.equals(ele)).forEach(ele->unsubscribeTopic(ele,baseSocketSession));
        }

    }

    public static  Collection<BaseSocketSession> getAllSessionsByTopic(String topic){
        Set<BaseSocketSession> baseSocketSessions = topicsChannel.get(topic);
        return CollectionUtils.isEmpty(baseSocketSessions) ? CollectionUtils.EMPTY_COLLECTION:baseSocketSessions;
    }

    public static void unsubscribeTopic(String topic, BaseSocketSession baseSocketSession) {
        Set<BaseSocketSession> baseSocketSessions = topicsChannel.get(topic);
        if (CollectionUtils.isNotEmpty(baseSocketSessions)) {
            baseSocketSessions.remove(baseSocketSession);
            if (CollectionUtils.isEmpty(baseSocketSessions)){
                topicsChannel.remove(topic);
            }
        }

        Set<String> existTopics = channelTopics.get(baseSocketSession);
        if (CollectionUtils.isNotEmpty(existTopics)){
            existTopics.remove(topic);
            if (CollectionUtils.isEmpty(existTopics)){
                channelTopics.remove(baseSocketSession);
            }
        }
    }

    public static void sendTopicMessage(String domain, String topic, Collection<String> filtered , int qos , RrmpMessage rrmpMessage, Message wsMessage) {
        if (StringUtil.isNullOrEmpty(topic)) {
            log.warn("send topic message ,but topic is null");
            return;
        }
        // copy a list
        if (topicsChannel.get(topic)==null) {
            log.warn("send topic,topic={} channel is null",topic);
            return;
        }
        Set<BaseSocketSession> socketSessions = Sets.newHashSet(topicsChannel.get(topic));
        if (CollectionUtils.isEmpty(filtered)){
            filtered= CollectionUtils.EMPTY_COLLECTION;
        }

        if (socketSessions != null && socketSessions.size() > 0) {
            Collection<String> finalFiltered = filtered;

            final Iterator<BaseSocketSession> iterator = socketSessions.iterator();
            while (iterator.hasNext()){
                final BaseSocketSession session = iterator.next();
                if (session.domain().equals(domain)&& !finalFiltered.contains(session.userId())) {
                    if (!session.channel().isActive()) {
                        iterator.remove();
                        unsubscribeTopic(topic,session);
                        continue;
                    }

                    topicPoolExecutor.submit(new SmartRunnable(qos) {
                        @Override
                        public void run() {
                            if (session.protocol() == ProtocolEnum.RRPM) {
                                session.sendMessage(rrmpMessage).addListeners((ChannelFutureListener) future -> {
                                    if (!future.isSuccess() && future.cause() instanceof ClosedChannelException) {
                                        unsubscribeTopic(topic,session);
                                    }else {
                                    }
                                });
                            } else {
                                if (wsMessage.getHeader().getSeq() == 0) {
                                    wsMessage.getHeader().setSeq(session.nextSeqId());
                                }
                                session.sendMessage(wsMessage).addListener((ChannelFutureListener) future -> {
                                    if (!future.isSuccess() && future.cause() instanceof ClosedChannelException) {
                                        unsubscribeTopic(topic,session);
                                    }else {
                                    }
                                });
                            }
                        }
                    });
                }
            }
        }
    }
}
